package cz.data.common.kafka.manage;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import cz.data.common.kafka.KafkaUtils;
import cz.data.common.support.db.DbProperty;
import cz.data.common.support.db.DbType;
import cz.data.common.support.db.core.DbColumn;
import cz.data.common.support.db.core.DbTable;
import cz.data.common.support.db.manage.Condition;
import cz.data.common.support.db.manage.DbManage;
import cz.data.common.support.metadata.JavaTypeEnum;
import cz.data.common.utils.ThrowableUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class KafkaDbManage implements DbManage {
    private static final String DEFAULT_GROUP = "java_kafka_driver";
    private final DbProperty property;
    private AdminClient adminClient;
    private KafkaConsumer<String, String> consumer;

    public KafkaDbManage(DbProperty property) {
        this.property = property;
    }

    private synchronized void adminClientInitialization() {
        if (Objects.isNull(this.adminClient)) {
            String bootstrapServer = DbType.KAFKA.getUrl()
                    .replace("${host}", property.getHost())
                    .replace("${port}", Optional.ofNullable(property.getPort())
                            .map(String::valueOf).orElse("9092"));
            Map<String, Object> configure = KafkaUtils.defaultAdminClientConfigure(bootstrapServer);
            this.adminClient = AdminClient.create(configure);
        }
    }

    private synchronized void consumerClientInitialization(String groupId) {
        if (Objects.isNull(this.consumer)) {
            String bootstrapServer = DbType.KAFKA.getUrl()
                    .replace("${host}", property.getHost())
                    .replace("${port}", Optional.ofNullable(property.getPort())
                            .map(String::valueOf).orElse("9092"));
            Map<String, Object> configure = KafkaUtils.defaultConsumerConfigure(bootstrapServer);
            configure.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
            this.consumer = new KafkaConsumer<>(configure);
        }
    }

    @Override
    public List<DbTable> getTables(String dbName, String schemaName) {
        this.adminClientInitialization();
        ListTopicsResult listTopicsResult = this.adminClient.listTopics();
        Set<String> topicNames = ThrowableUtil.callable(() -> listTopicsResult.names().get(30, TimeUnit.SECONDS));
        if (CollectionUtils.isEmpty(topicNames))
            return Collections.emptyList();
        List<DbTable> tables = new ArrayList<>();
        for (String topicName : topicNames) {
            if ("__consumer_offsets".equals(topicName))
                continue;
            DbTable table = new DbTable();
            table.setTableName(topicName);
            table.setTableComment(topicName);
            tables.add(table);
        }
        return tables;
    }

    @Override
    public List<DbColumn> getTableColumns(String dbName, String schemaName, String tableName) {
        this.consumerClientInitialization(DEFAULT_GROUP);
        consumer.subscribe(Collections.singletonList(tableName));
        consumer.poll(0);
        Set<TopicPartition> assignment = consumer.assignment();
        // 重置偏移量至末尾
        consumer.seekToEnd(assignment);
        // 提交偏移量
        Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(assignment);
        for (Map.Entry<TopicPartition, OffsetAndMetadata> entry : committed.entrySet()) {
            OffsetAndMetadata value = entry.getValue();
            //计算末端偏移量-1获取以最后一条数据
            final long offset = value.offset() > 0 ? value.offset() - 1 : 0;
            consumer.seek(entry.getKey(), offset);
        }
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(3));
        if (records.isEmpty())
            return Collections.emptyList();
        List<DbColumn> columns = new ArrayList<>();
        for (ConsumerRecord<String, String> record : records) {
            String value = record.value();
            JSONObject jsonObject = ThrowableUtil.callable(() -> JSONObject.parseObject(value), false);
            if (Objects.isNull(jsonObject))
                continue;
            int i = 1;
            for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
                String columnName = entry.getKey();
                Object columnValue = entry.getValue();
                String type;
                if (columnValue instanceof Map) {
                    type = JavaTypeEnum.OBJECT.getValue();
                } else if (columnValue instanceof Collection) {
                    type = JavaTypeEnum.ARRAY.getValue();
                } else {
                    type = JavaTypeEnum.match(value.getClass().getSimpleName()).getValue();
                }
                DbColumn column = new DbColumn();
                column.setColName(columnName);
                column.setDataType(type);
                column.setColPosition(i++);
                column.setColComment(columnName);
                column.setNullable(false);
                column.setColKey(false);
                columns.add(column);
            }
            break;
        }
        return columns;
    }

    @Override
    public boolean createTable(DbTable table, Properties properties) {
        this.adminClientInitialization();
        String partitionNum = properties.getProperty("topic.partitions", "1");
        short r = 0;
        NewTopic topic = new NewTopic(table.getTableName(),
                NumberUtils.toInt(partitionNum, 1), r);
        CreateTopicsResult result = this.adminClient.createTopics(Collections.singletonList(topic));
        try {
            result.all().get(3, TimeUnit.SECONDS);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    @Override
    public boolean deleteTable(DbTable table) {
        this.adminClientInitialization();
        DeleteTopicsResult result = this.adminClient.deleteTopics(Collections.singletonList(table.getTableName()));
        try {
            result.all().get(3, TimeUnit.SECONDS);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    @Override
    public boolean insertData(DbTable table, JSONObject record) {
        return false;
    }

    @Override
    public boolean updateData(DbTable table, JSONObject record) {
        return false;
    }

    @Override
    public boolean removeData(DbTable table, JSONObject record) {
        return false;
    }

    @Override
    public IPage<JSONObject> getPageData(DbTable table, List<String> fields, List<Condition> conditions, Page<JSONObject> page) {
        return page;
    }

    @Override
    public JSONObject getDataById(DbTable table, JSONObject params) {
        return null;
    }

    @Override
    public void close() throws Exception {
        if (Objects.nonNull(this.adminClient)) {
            this.adminClient.close();
        }
        if (Objects.nonNull(this.consumer)) {
            this.consumer.close();
        }
    }
}
